package com.ds.lens.flink.monitor.monitorjob.function;

import com.ds.lens.data.common.client.LensResponseData;
import com.ds.lens.data.common.client.vo.LensMonitorStateSaveInVO;
import com.ds.lens.data.common.constant.MonitorStateEnum;
import com.ds.lens.data.common.document.MetricState;
import com.ds.lens.data.common.document.dto.MonitorEvent;
import com.ds.lens.flink.monitor.MonitorJobContext;
import com.ds.lens.flink.monitor.common.BuilderUtil;
import com.ds.lens.flink.monitor.common.LogUtils;
import com.ds.lens.flink.monitor.common.MonitorDefinitionFactory;
import com.ds.lens.flink.monitor.common.bo.Metric;
import com.ds.lens.flink.monitor.common.bo.MetricBundle;
import com.ds.lens.flink.monitor.common.bo.MonitorDefinition;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Description:
 *
 * @author WeiShaoying
 * @date 2020-05-13
 */
public class UpdateEventStateFunction extends ProcessFunction<Tuple2<MetricBundle, Map<String, MonitorEvent>>, MonitorEvent> {
    private static final long serialVersionUID = 1815169085541081442L;

    @Override
    public void processElement(Tuple2<MetricBundle, Map<String, MonitorEvent>> tuple2, Context context, Collector<MonitorEvent> out) throws Exception {
        MetricBundle metricBundle = tuple2.f0;
        Map<String, MonitorEvent> eventMap = tuple2.f1;
        MonitorDefinition definition = MonitorDefinitionFactory.get().getMonitorDefinitionByState(metricBundle.getState());
        LogUtils.debug("Input param metric bundle {}", metricBundle);
        if (definition == null || !definition.getMonitorState().getKey().getVersion().equals(metricBundle.getState().getKey().getVersion())) {
            return;
        }
        List<MonitorEvent> events = updateEventState(definition, eventMap, metricBundle);
        for (MonitorEvent event : events) {
//            LogUtils.debug("collect event={}", event);
//            System.out.println("send event="+new Gson().toJson(event));
            out.collect(event);
        }
    }

    private List<MonitorEvent> updateEventState(MonitorDefinition definition, Map<String, MonitorEvent> eventMap, MetricBundle bundle) {
        Map<String, Integer> targetStateMap = new HashMap<>();
        for (MonitorEvent evt : eventMap.values()) {
            targetStateMap.put(evt.getTriggered().getGroupValue(), evt.getTriggered().getState().getState());
        }

        LensMonitorStateSaveInVO inVO = LensMonitorStateSaveInVO.builder()
                .monitorId(definition.getMonitorState().getKey().getMonitorId())
                .targets(targetStateMap)
                .version(definition.getMonitorState().getKey().getVersion())
                .missing(definition.getNotification().getNotifyNoData())
                .mute(definition.getNotification().getIsMute())
                .interval(definition.getIntervalTime())
                .modifiedTime(bundle.getWindowEndTime())
                .renotifyDuration(definition.getNotification().getRenotifyDuration())
                .build();
        LensResponseData<Map<String, MetricState>> resp = MonitorJobContext.getInstance().getMonitorClient().saveMonitorState(inVO);
        LogUtils.debug("send LensMonitorStateSaveInVO={} and resp={}, bundle={}", inVO, resp, bundle);
        List<MonitorEvent> list = new ArrayList<>();
        if (resp != null) {
            Map<String, MetricState> dataMap = resp.getData();
            if (!CollectionUtils.isEmpty(dataMap)) {
                for (String target : dataMap.keySet()) {
                    MetricState metricState = dataMap.get(target);
                    MonitorStateEnum state = MonitorStateEnum.valueOf(metricState.getState());
                    MonitorEvent event = eventMap.get(target);
                    if (event != null) {
                        event.getTriggered().setState(state);
                    } else {
                        Map<String, Object> groupBys = BuilderUtil.parseGroupValue(target);
                        Metric metric = BuilderUtil.buildMetric(definition.getMonitorState().getKey().getMonitorId(),
                                definition.getMonitorState().getKey().getVersion(), definition.getMetricName(),
                                definition.getMetricAggr(), groupBys, null, bundle.getStartTime(), bundle.getEndTime());
                        event = BuilderUtil.buildMonitorEvent(definition, metric, null, null, state);
                    }
                    event.getTriggered().setRenotify(metricState.getRenotify() == null ? false : metricState.getRenotify());
                    list.add(event);
                }
            }
        }
        return list;
    }
}
